package cn.chencq.proxy.timer.service;

import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cn.chencq.proxy.constants.RedisKey;
import cn.chencq.proxy.util.Util;

import com.limei.adplatform.redis.cluster.RedisClusterFactory;

public class ProxyCheckService {
	
	private static final Logger LOGGER = LoggerFactory.getLogger(ProxyIPService.class);
	
	private static ExecutorService threadPool = 
			new ThreadPoolExecutor(
							200, 
							200,
							5 * 1000, 
							TimeUnit.MINUTES, 
							new LinkedBlockingDeque<Runnable>(), 
							new ThreadCreateFactory());

	static class ThreadCreateFactory implements java.util.concurrent.ThreadFactory {

		final ThreadGroup group;
		final AtomicInteger threadNumber = new AtomicInteger(1);
		final String namePrefix;

		public ThreadCreateFactory() {
			SecurityManager s = System.getSecurityManager();
			group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
			namePrefix = "pool-thread-check";
		}

		@Override
		public Thread newThread(Runnable r) {
			Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(), 0);
			if (t.isDaemon())
				t.setDaemon(false);
			if (t.getPriority() != Thread.NORM_PRIORITY)
				t.setPriority(Thread.NORM_PRIORITY);
			return t;
		}

	}

	public void check() {
		String clusterName = RedisKey.getClusterName();
		String key = RedisKey.getProxyQueueKey();

		Long beforeLen = RedisClusterFactory.llen(clusterName, key);
		
		if(beforeLen <= 0){
			LOGGER.info("current proxy queue is empty");
			return;
		}
		
		LOGGER.info("before check proxy, queue length:{}",beforeLen);
		int pageSize = 100;
		int totalPage = beforeLen.intValue() / pageSize + 1;

		int end = pageSize - 1;
		int start = 0;
		
		int count = 0;
		Set<Future<Integer>> sfSet = new HashSet<Future<Integer>>();
		
		for(int page = 1; page <= totalPage ; page++){
			
			start = pageSize *(page-1);
			
			end = pageSize*page-1;
					
			List<String> list = RedisClusterFactory.lrange(clusterName, key, start, end);
			Set<String> pSet = Util.listToSet(list);
			
			sfSet.add(threadPool.submit(new ProxyCheckCallable(pSet,page)));
		}

		for (Future<Integer> future : sfSet) {
			try {
				Integer result = future.get();
				count += result;
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
		
		Long curLen = RedisClusterFactory.llen(clusterName, key);
		LOGGER.info("after check proxy, queue length:{}",curLen);
		
		//移除已经校验的数据
		if(curLen > beforeLen){
			RedisClusterFactory.ltrim(clusterName, key,beforeLen.intValue(),curLen.intValue());
		}else{
			RedisClusterFactory.del(clusterName, key);
		}
		
		LOGGER.info("current check remove igel proxy num:" + count);
	}

	
	public static void main(String[] args) {
		RedisClusterFactory.init("redis_7379.xml");
		new ProxyCheckService().check();
	}
}
